package com.hmall.search.listener;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ObjUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import com.hmall.api.client.ItemClient;
import com.hmall.api.dto.ItemDTO;
import com.hmall.api.dto.ItemDoc;
import com.hmall.common.enums.SearchStatus;
import com.hmall.search.utils.EsUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.BoundValueOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
/**
 * @author zhangyan
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class SearchMessageListener {
    private final ItemClient itemClient;
    private final EsUtil esUtil;
    private final RedisTemplate<String,String> redisTemplate;
    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue(name = SearchStatus.QUEUE_ITEM_ADD),
                    exchange = @Exchange(name = SearchStatus.EXCHANGE_ITEM_DIRECT, delayed = "true"),
                    key = SearchStatus.RK_ITEM_ADD
            ),
            @QueueBinding(
                    value = @Queue(name = SearchStatus.QUEUE_ITEM_UPDATE),
                    exchange = @Exchange(name = SearchStatus.EXCHANGE_ITEM_DIRECT, delayed = "true"),
                    key = SearchStatus.RK_ITEM_UPDATE
            )
    })
    public void listenSearchAddOrUpdateMessage(Message message) {
        String messageId = message.getMessageProperties().getMessageId();
        String bodyStr = new String(message.getBody(), StandardCharsets.UTF_8);
        Long itemId = Long.valueOf(bodyStr);
        if (!isValidItemId(itemId)) return;

        String redisKey = "search-service:item:addOrUpdate:" + messageId;
        if (checkAndMarkProcessed(redisKey)) {
            log.info("【ES监听】消息已处理，跳过，messageId = {}", messageId);
            return;
        }

        log.info("【ES监听】接收到新增/修改消息，商品ID：{}", itemId);
        syncItemToES(itemId);
    }

    @RabbitListener(bindings =
    @QueueBinding(
            value = @Queue(name = SearchStatus.QUEUE_ITEM_DELETE),
            exchange = @Exchange(name = SearchStatus.EXCHANGE_ITEM_DIRECT, delayed = "true"),
            key = SearchStatus.RK_ITEM_DELETE
    ))
    public void listenSearchDeleteMessage(Message message) {
        String messageId = message.getMessageProperties().getMessageId();
        String bodyStr = new String(message.getBody(), StandardCharsets.UTF_8);
        Long itemId = Long.valueOf(bodyStr);
        if (!isValidItemId(itemId)) return;


        String redisKey = "search-service:item:delete:" + messageId;
        if (checkAndMarkProcessed(redisKey)) {
            log.info("【ES监听】消息已处理，跳过，itemId = {}", messageId);
            return;
        }

        log.info("【ES监听】接收到删除消息，商品ID：{}", itemId);
        try {
            esUtil.deleteDoc(itemId.toString());
            log.info("【ES监听】已成功从ES删除商品，商品ID：{}", itemId);
        } catch (IOException e) {
            log.error("【ES监听】从ES删除商品失败，商品ID：{}", itemId, e);
        }
    }

    // === 抽取共用方法 ===

    private boolean checkAndMarkProcessed(String redisKey) {
        BoundValueOperations<String, String> ops = redisTemplate.boundValueOps(redisKey);
        if (StrUtil.isNotBlank(ops.get())) {
            return true;
        }
        int expire = 60 + RandomUtil.randomInt(0, 60);
        ops.set("processed", expire, TimeUnit.SECONDS);
        return false;
    }

    private boolean isValidItemId(Long itemId) {
        if (ObjUtil.isEmpty(itemId)) {
            log.warn("【ES监听】接收到空的商品ID，忽略处理");
            return false;
        }
        return true;
    }

    private void syncItemToES(Long itemId) {
        ItemDTO itemDTO = itemClient.queryItemById(itemId);
        if (ObjUtil.isEmpty(itemDTO)) {
            log.warn("【ES监听】未查询到商品信息，ID：{}", itemId);
            return;
        }
        ItemDoc itemDoc = BeanUtil.toBean(itemDTO, ItemDoc.class);
        try {
            esUtil.addDoc(itemDoc);
            log.info("【ES监听】已成功同步商品到ES，商品ID：{}", itemId);
        } catch (IOException e) {
            log.error("【ES监听】同步商品到ES失败，商品ID：{}", itemId, e);
        }
    }
}
